package retail.loader;

import java.math.BigDecimal;
import java.math.BigInteger;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;

import org.apache.log4j.LogManager;
import org.springframework.jdbc.core.BatchPreparedStatementSetter;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.util.StringUtils;

public class Worker implements Runnable {
	static final org.apache.log4j.Logger logger2 = LogManager
			.getLogger(Worker.class);
	private final CountDownLatch mDoneSignal;
	private final int pageIndex;
	private String tableName;
	private final TableLoader tableLoader;
	SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

	public Worker(final TableLoader tableLoader,
			final CountDownLatch doneSignal, final int pageIndex) {
		this.mDoneSignal = doneSignal;
		this.pageIndex = pageIndex;
		this.tableName = tableLoader.getTableName();

		this.tableLoader = tableLoader;
	}

	public void run() {
		long take = System.currentTimeMillis();
		try {
			int result = LoadData();

			logger2.info("loaded " + tableName + " at page index " + pageIndex
					+ " rows " + result + " take + "
					+ (System.currentTimeMillis() - take) + " ms. ");

		} catch (Exception e) {
			logger2.error(e);
			tableLoader.haserror = true;
		}
		mDoneSignal.countDown();
	}

	private int LoadData() throws Exception {

		Connection con = Loader.getMysqlConnection();
		String where = "";
		if (StringUtils.hasLength(tableLoader.maxId)) {
			if (null != Loader.updateDateMin) {
				where = " WHERE id >='" + tableLoader.maxId + "'";
				where += " OR ( update_time <='"
						+ formatter.format(Loader.updateDateMax) + "' ";
				where += " AND update_time>='"
						+ formatter.format(Loader.updateDateMin) + "' )";
			} else
				where = " WHERE id <='" + tableLoader.maxId + "' ";
		} else {
			where = " WHERE update_time <='"
					+ formatter.format(Loader.updateDateMax) + "' ";
			if (null != Loader.updateDateMin) {
				where += " AND update_time>='"
						+ formatter.format(Loader.updateDateMin) + "'";
			}
		}
		
		
//		if (tableLoader.isSharding
//				&& !StringUtils.hasLength(Loader.shardingFlag)) {
//			where += " AND sharding_flag = '" + Loader.shardingFlag + "'";
//		}
		

		try {
			String script = String.format("Select %s From %s ",
					tableLoader.fieldString, tableName)
					+ where
					+ " order by id limit "
					+ pageIndex
					* tableLoader.pageSize
					+ " , " + tableLoader.pageSize;
			logger2.debug(script);
			PreparedStatement ps = con.prepareStatement(script);
			ps.setFetchSize(10);
			ps.execute();
			ResultSet rs = ps.getResultSet();
			int result = 0;
			if (null == Loader.updateDateMin)
				result = excute(rs);
			else
				result = merge(rs);

			tableLoader.count(result);

			if (result < tableLoader.pageSize) {
				this.tableLoader.finish(true);
			}
			rs.close();
			con.close();
			return result;
		} catch (Exception e) {
			logger2.error("load error", e);
			con.close();
			tableLoader.haserror = true;
			tableLoader.finished = true;
			Loader.errors.put(tableName, e.getMessage());
			return 0;
		}

	}

	// List<String> _fields;

	private Map<String, Integer> getFeilds(ResultSetMetaData meta)
			throws Exception {
		// _fields = new ArrayList<>();
		int count = meta.getColumnCount();
		HashMap<String, Integer> map = new HashMap<>();
		for (int i = 1; i <= count; i++) {
			String name = meta.getColumnName(i).toUpperCase();
			map.put(name, i);
		}
		return map;
	}

	private int excute(ResultSet reader) throws Exception {
		Connection con = Loader.getOracleConnection();

		int mode = 5000;
		PreparedStatement statement;
		ResultSetMetaData meta = reader.getMetaData();
		Map<String, Integer> fieldMap = getFeilds(meta);
		String fields = "";
		String values = "";
		int v = 1;
		for (String key : tableLoader.fieldTypes.keySet()) {
			if (v > 1) {
				fields += ",";
				values += ",";
			}
			fields += ("\"" + key.toUpperCase() + "\"");
			values += "?";
			v += 1;
		}

		String script = "INSERT INTO " + tableLoader.descTable.toUpperCase()
				+ "(" + fields + ") VALUES (" + values + ")";
		logger2.debug(script);
		int len = 0;
		statement = con.prepareStatement(script);
		long take = System.currentTimeMillis();
		while (reader.next()) {
			if (len == 0) {
				logger2.debug("begin get data takes:"
						+ (System.currentTimeMillis() - take));
			}
			v = 1;
			for (String key : tableLoader.fieldTypes.keySet()) {				
<<<<<<< HEAD
				String type = getFieldType(key); 
				Object val = reader.getObject(key);;
				if( key.equalsIgnoreCase("status") || ( "number".equalsIgnoreCase(type) && val instanceof Boolean) )
					val = reader.getInt(key); 
				
				//reader.getMetaData().getColumnName(80);
=======
				//int index = fieldMap.get(key);
				Object val = reader.getObject(key);
				if(key.equalsIgnoreCase("status"))
					val = reader.getInt(key);
				
				reader.getMetaData().getColumnName(80);
>>>>>>> 1ce2a0af3d001f8d009b9ae2a3a38d3be863296c
				setStatementValue(statement, v, val, key);
				v += 1;
			}
			statement.addBatch();
			len += 1;
			if ((len % mode) == 0) {
				try {
					statement.executeBatch();
				} catch (Exception e) {
					logger2.debug("error in " + len + " at " + pageIndex);
					throw e;
				}

				statement.close();
				statement = con.prepareStatement(script);
			}
		}
		if (len == 0) {
			statement.close();
			con.close();
			return 0;
		}
		if ((len % mode) != 0)
			statement.executeBatch();

		statement.close();
		con.close();
		return len;
	}

	private String getFieldType(String field) {
		String type = "";
		if (tableLoader.fieldTypes.containsKey(field)) {
			type = tableLoader.fieldTypes.get(field);
		}
		return type;
	}

	private boolean setStatementValue(PreparedStatement statement, int index,
			Object val, String field) throws SQLException {
		String type = getFieldType(field);
<<<<<<< HEAD
		
=======
>>>>>>> 1ce2a0af3d001f8d009b9ae2a3a38d3be863296c
		int i = index;
		if (val instanceof Integer)
			statement.setInt(i, (Integer) val);
		else if (val instanceof Double)
			statement.setDouble(i, (Double) val);
		else if (val instanceof Timestamp)
			statement.setTimestamp(i, (java.sql.Timestamp) val);
		else if (val instanceof Long)
			statement.setLong(i, (Long) val);
		else if (val instanceof BigInteger)
			statement.setLong(i, Long.parseLong(val.toString()));
		else if (val instanceof Short)
			statement.setShort(i, (Short) val);
		else if (val instanceof Boolean)
			statement.setBoolean(i, (boolean) val);
		else if (val instanceof Byte)
			statement.setByte(i, (byte) val);
		else if (val instanceof String)
			statement.setString(i, (String) val);
		else if (val instanceof BigDecimal)
			statement.setBigDecimal(i, (BigDecimal) val);
		else {
			if (type.equalsIgnoreCase("NUMBER")) {
				if (val == null)
					val = 0;
				statement.setInt(i, (Integer) val);
			} else if (type.equalsIgnoreCase("DATE"))
				statement.setDate(i, (java.sql.Date) val);
			else if (type.equalsIgnoreCase("VARCHAR2"))
				statement.setString(i, (String) val);
			else
				statement.setObject(i, val);
		}

		return true;
	}

	private int merge(ResultSet reader) throws Exception {

		String sql = getMergeScript(reader);
		logger2.debug(sql);

		ResultSetMetaData meta = reader.getMetaData();
		Map<String, Integer> fieldMap = getFeilds(meta);
		int count = meta.getColumnCount();
		List<Map<String, Object>> lst = new ArrayList<>();
		int len = 0;
		while (reader.next()) {
			HashMap<String, Object> vals = new HashMap<>();
			for (int i = 1; i <= count; i++) {
<<<<<<< HEAD
				String key = meta.getColumnName(i).toUpperCase(); 
				String type = getFieldType(key); 
				Object val =  reader.getObject(i);;
				if( key.equalsIgnoreCase("status") || ( "number".equalsIgnoreCase(type) && val instanceof Boolean) )
					val = reader.getInt(i);	 
				
=======
				String key = meta.getColumnName(i).toUpperCase();
				Object val = reader.getObject(i);
				if( key.equalsIgnoreCase("status"))
					val = reader.getInt(i);				
>>>>>>> 1ce2a0af3d001f8d009b9ae2a3a38d3be863296c
				vals.put(key, val);
			}
			lst.add(vals);
			len += 1;
			if ((len % 5000) == 0) {
				batchExcute(sql, lst);
				lst = new ArrayList<>();
			}
		}

		if ((len % 5000) != 0) {
			batchExcute(sql, lst);
		}

		return len;
	}

	private void batchExcute(String sql, List<Map<String, Object>> lst) {
		JdbcTemplate template = (JdbcTemplate) Loader.context
				.getBean("jdbcTemplate");
		BatchPreparedStatementSetter setter = getBatchPreparedStatementSetter(
				lst, columnNames);
		template.batchUpdate(sql, setter);

	}

	private BatchPreparedStatementSetter getBatchPreparedStatementSetter(
			final List<Map<String, Object>> lst, final String[] fields) {
		return new BatchPreparedStatementSetter() {

			public void setValues(PreparedStatement ps, int idx)
					throws SQLException {
				Map<String, Object> vals = lst.get(idx);

				for (int i = 0; i < fields.length; i++) {
					String field = fields[i];
					Object val = vals.get(field);
					setStatementValue(ps, i + 1, val, field);
				}
				setStatementValue(ps, fields.length + 1, vals.get("ID"), "id"); // ID
																				// 列单独设置
			}

			public int getBatchSize() {
				return lst.size();
			}
		};
	}

	String[] columnNames;

	private String getMergeScript(ResultSet reader) throws Exception {
		ResultSetMetaData meta = reader.getMetaData();
		columnNames = new String[tableLoader.fieldTypes.size() - 1]; // 排除ID列
		// for (int i = 0; i < columnNames.length; i++) {
		// columnNames[i] = meta.getColumnName(i + 2);
		// }

		int i = 0;
		for (String string : tableLoader.fieldTypes.keySet()) {
			if (string.equalsIgnoreCase("id"))
				continue;
			columnNames[i++] = string;
		}
		String[] viewColumnNames = new String[] {};
		OracleSqlTemplate template = new OracleSqlTemplate();
		String[] keys = new String[] { "ID" };
		String script = template.getMergeSql(Config.current.getOracleUsr(),
				tableLoader.tableName, keys, columnNames, viewColumnNames);
		return script;
	}
}
